package dwetl.transformer;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCreator;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.KeyHolder;
import org.springframework.jdbc.support.rowset.SqlRowSet;
import org.springframework.stereotype.Service;

import dwetl.cache.TablesUkCache;
import dwetl.model.DimColumn;
import dwetl.model.DimMdfTable;
import dwetl.model.DimTable;
import dwetl.util.ConfigUtil;

@Service
@Scope("prototype")
public class DimMdfTalbeTransformer {
    private static final Logger logger = LoggerFactory.getLogger(DimMdfTalbeTransformer.class);

    @Autowired
    private JdbcTemplate jdbcTemplate;

    private TablesUkCache tablesUkCache;

    public void enableDimCache(boolean enable){
        if(enable){
            tablesUkCache=new TablesUkCache();
        }else{
            tablesUkCache=null;
        }
    }

    public Object doTransform(DimMdfTable dimMdfTable , Map<String,Object> rowMap){
        Map<String,Map<String,Object>> ctxMap=new HashMap<String,Map<String,Object>>();

        List<DimTable> dimTables = dimMdfTable.getDimTables();
        for(DimTable dt:dimTables){
            putDimTable(dt,rowMap,ctxMap);
        }
        return putDimTable(dimMdfTable,rowMap,ctxMap);
    }

    Object putDimTable(DimTable dt,Map<String,Object> rowMap,Map<String,Map<String,Object>> ctxMap){
//        String table=dt.getTable_name();
        DimColumn autoIdCol=dt.findAutoIdColumn();

        if(autoIdCol==null){
            throw new RuntimeException(dt.getTableWithAlias()+"配置错误，缺少is_auto_id：True的列");
        }

        SortedSet<DimColumn> uks= dt.getUnique_key_column_set();

        boolean hasSame=false;
        Map<String,Object> thtMap=null;
        if(this.tablesUkCache!=null){
            Object[] hasCacheRl=tablesUkCache.hasSameUkCache(dt,uks,rowMap,ctxMap);
            hasSame=Boolean.valueOf(""+hasCacheRl[0]);
            thtMap=(Map<String,Object>)hasCacheRl[1];
        }
        if(hasSame){//cache记录已存在
            ctxMap.put(dt.getTable_alias(),thtMap);
        }else if(checkDimExists(dt,autoIdCol.getCol(),rowMap,ctxMap)){//数据库表记录已存在
            if(dt.isIf_exists_skip()==false){
                return insertDim(dt,autoIdCol.getCol(),rowMap,ctxMap);
            }
        }else{
            return insertDim(dt,autoIdCol.getCol(),rowMap,ctxMap);
        }
        return null;
    }

    boolean checkDimExists(DimTable dt,String autoIdCol,Map<String,Object> rowMap,Map<String,Map<String,Object>> ctxMap){
        String and_where="";
        List<Object> params=new LinkedList<Object>();
        SortedSet<String> uks= dt.getUnique_key_columns();
        if(uks==null || uks.size()<1){//没有唯一键
            return false;
        }
        for(String col:uks){//unique_key_columns拼装查询条件
            try {
                DimColumn dc = dt.findDimColumnByCol(col);
                Object srcColVal = rowMap.get(dc.getSrc_col());
                if (dc.getDim_id_from() != null && dc.getDim_id_from().length() > 0) {
                    srcColVal = this.getDimIdFrom(dc.getDim_id_from(), ctxMap);
                }
                if (srcColVal == null) {
                    and_where += " and " + dc.getCol() + " is null ";
                } else if (srcColVal instanceof java.util.Date) {
                    and_where += " and DATE_FORMAT(" + dc.getCol() + ",'%Y-%m-%d %H:%i:%s')=? ";
                    params.add(ConfigUtil.formatDateTime((java.util.Date) srcColVal));
                } else {
                    and_where += " and " + dc.getCol() + "=? ";
                    params.add(srcColVal);
                }
            }catch(Exception e){
                throw new RuntimeException("Failed with col : "+col,e);
            }
        }
        String sql=String.format("select * from %s where 1=1 %s order by %s desc limit 1",dt.getTable_name(),and_where,autoIdCol);
        logger.info("query sql:"+sql +" params:"+params);

        Map<String,Object> tblMap=new HashMap<String,Object>();//行值map
        SqlRowSet result = jdbcTemplate.queryForRowSet(sql,params.toArray());
        if (result.next()) {//根据唯一性字段找到对应的行，检查各个字段是否有改动
            for(DimColumn dc:dt.getEtl_columns()){
                Object valFromTbl=result.getObject(dc.getCol());
                if(dt.getUncheck_columns().contains(dc.getCol()) || dc.getCol().equals(dt.getUnique_key_date_column())){
                    //如果uncheck_columns包含了该列则不需要判断是否值有改动过
                    //如果是unique_key_date_column则不需要判断是否值有改动过
                    tblMap.put(dc.getCol(),valFromTbl);
                    continue;
                }else if(dc.getDim_id_from()!=null && dc.getDim_id_from().length()>0){
                    tblMap.put(dc.getCol(),valFromTbl);
                    continue;
                }else if(autoIdCol.equals(dc.getCol())){//如果是自增id列
                    tblMap.put(autoIdCol,valFromTbl);
                    continue;
                }else{
                    Object valFromNew=rowMap.get(dc.getSrc_col());
                    tblMap.put(dc.getCol(),valFromTbl);
                    if(valFromNew instanceof java.util.Date) {
                        java.util.Date newDate=(java.util.Date)valFromNew;
                        java.util.Date tblDate=(java.util.Date)valFromTbl;
                        if(newDate.getTime()==tblDate.getTime()){
                            continue;
                        }else{
                            return false;
                        }
                    }else if(valFromTbl instanceof Number){
                        double newDouble=Double.parseDouble(valFromNew.toString());
                        double tblDouble=((Number)valFromTbl).doubleValue();
                        if(newDouble==tblDouble){
                            continue;
                        }else{
                            return false;
                        }
                    }else if((""+valFromTbl).equals((""+valFromNew))){
                        continue;
                    }else{
                        return false;
                    }
                }
            }
//            result
        }else{//根据唯一性字段找不到对应的行
            return false;
        }

        //到这里来了就代表同样的行已经存在了，把改行值保存到ctxMap待用
        ctxMap.put(dt.getTable_alias(),tblMap);
        //put cache
        if(tablesUkCache!=null){
            tablesUkCache.putCache(dt,rowMap,tblMap);
        }
        return true;
    }

    private Object getDimIdFrom(String didf,Map<String,Map<String,Object>> ctxMap){
        try {
            String dimTbl = didf.split("\\.")[0];
            String dimTblCol = didf.split("\\.")[1];
            return ctxMap.get(dimTbl).get(dimTblCol);
        }catch(Exception e){
            logger.warn(didf+" 配置有误，建议检查。");
            return null;
        }
    }

    private Object insertDim(DimTable dt,String autoIdCol,Map<String,Object> rowMap,Map<String,Map<String,Object>> ctxMap){
//        String table=dt.getTable_name();
        List<String> insCols=new LinkedList<String>();
        List<String> wenhaos=new LinkedList<String>();
        List<Object> insColObjs=new LinkedList<Object>();
        Object providedId=null;
        for(DimColumn dc:dt.getEtl_columns()){
            if(dc.getIs_auto_id() && ("".equals(dc.getSrc_col()) || dc.getSrc_col()==null)){//配置了is_auto_id同时也配了src_col则代表用户不使用自增键而使用自己提供的键，数据去重有效性用户自己承担。
                continue;
            }else if(dc.getIs_auto_id()){
                providedId=rowMap.get(dc.getSrc_col());
            }
            // String tblCol=dc.getCol();
            insCols.add(dc.getCol());
            wenhaos.add("?");
            String srcCol=dc.getSrc_col();
            if(null==dc.getDim_id_from() || dc.getDim_id_from().length()==0){//没有设置dim_id_from
                insColObjs.add(dc.valTransform(rowMap.get(srcCol)));
            }else{//有设置dim_id_from从维表拿
                Object val=getDimIdFrom(dc.getDim_id_from(),ctxMap);
                insColObjs.add(val);
            }
        }
        String insColStr=String.join(",",insCols);
        String wenhaoStr=String.join(",",wenhaos);
        String sql=String.format("insert into %s (%s) values (%s)",dt.getTable_name(),insColStr,wenhaoStr);
        logger.info("insert sql:"+sql);
        KeyHolder keyHolder = new GeneratedKeyHolder();
        jdbcTemplate.update( new PreparedStatementCreator(){
            @Override
            public PreparedStatement createPreparedStatement(Connection conn) throws SQLException{
                PreparedStatement ps = conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
                for(int i=0;i<insColObjs.size();i++) {
                    Object val = insColObjs.get(i);
                    if(val==null){
                        ps.setObject(i+1,null);
                    }else if(val instanceof java.util.Date){
                        java.sql.Timestamp dvl=new java.sql.Timestamp(((java.util.Date) val).getTime());
                        ps.setTimestamp(i + 1,dvl);
                    }else if (val instanceof java.sql.Date) {
                        ps.setTimestamp(i + 1, new java.sql.Timestamp(((java.sql.Date)val).getTime()));
                    }else{
                        ps.setString(1+i,""+val);
                    }
                }
                return ps;
            }
        },keyHolder);
        Number autoid= keyHolder.getKey();

        //put ctxMap
        Map<String,Object> tblMap=new HashMap<String,Object>();
        for(int i=0;i<insCols.size();i++){
            String key=insCols.get(i);
            Object val=insColObjs.get(i);
            tblMap.put(key,val);
        }
        tblMap.put(autoIdCol,autoid!=null?autoid:providedId);
        ctxMap.put(dt.getTable_alias(),tblMap);

        //put cache
        if(tablesUkCache!=null){
            tablesUkCache.putCache(dt,rowMap,tblMap);
        }

        return autoid;
    }
}
